package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by Smexy on 2023/4/6
 *
 *  把每种传感器最新的信息写入到Mysql。
 *
 *      不支持 EOS。
 *          原因： 不支持分布式事务。
 */
public class Demo4_JDBCSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
            SinkFunction<T> sink(
            String sql: 写出的sql语句。可以使用幂等输出的形式。
                    REPLACE INTO ws VALUES('s1',1,100);

                    INSERT INTO ws VALUES('s1',2,101)
                    ON DUPLICATE KEY UPDATE ts=VALUES(ts), vc = VALUES(vc);

            JdbcStatementBuilder<T> statementBuilder: 负责将流中数据的值，填充到PrepareStatement上。
                                                    填充占位符
            JdbcExecutionOptions executionOptions: 执行的一些参数设置。
                                                设置一批攒够多少条，写一次。
                                                设置每间隔多久写一次。

            JdbcConnectionOptions connectionOptions): 设置如何连接数据库
                                                            url,username,password
         */
        SinkFunction<WaterSensor> sink = JdbcSink
            .<WaterSensor>sink(
                "REPLACE INTO ws VALUES(?,?,?)",
                new JdbcStatementBuilder<WaterSensor>()
                {
                    /*
                        把 WaterSensor waterSensor相关的列值，填充到 占位符上。
                        占位符索引填充，从 1开始。
                     */
                    @Override
                    public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {
                        ps.setString(1, waterSensor.getId());
                        ps.setLong(2, waterSensor.getTs());
                        ps.setInt(3, waterSensor.getVc());
                    }
                },
                JdbcExecutionOptions.builder()
                                    .withMaxRetries(3) //写失败时，最多重试多少次。配置这个参数，最好使用幂等写出
                                    .withBatchSize(100) // 一批攒够多少条，发送一次写请求
                                    .withBatchIntervalMs(500) //长时间不够1批，每间隔500ms写一次
                                    .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                    .withUrl("jdbc:mysql://hadoop102:3306/221109?serverTimezone=Asia/Shanghai&useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true")
                    .withUsername("root")
                    .withPassword("000000")
                    .build()
            );

        env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   //old api
                   .addSink(sink);

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
